package cn.zx.rmq.producer;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.apache.rocketmq.remoting.exception.RemotingException;

import java.io.UnsupportedEncodingException;

/**
 * @Package: cn.zx.rmq.producer
 * @ClassName: AsyncProducer
 * @Author: Think
 * @CreateTime: 2021/6/28 11:06
 * @Description:不同的线程同时并发地进行消息的发送，比较适合各个消息之间无依赖的场景，发送效率得到了提升。
 */
@Slf4j
public class AsyncProducer {

    public static void main(String[] args) throws MQClientException, RemotingException, InterruptedException, UnsupportedEncodingException {
        DefaultMQProducer producer = new DefaultMQProducer("AsyncProducer-Group");
        producer.setNamesrvAddr("localhost:9876");
        producer.start();

        for (int i = 0; i < 5; i++) {
            Message msg = new Message("KYC", "about-health", ("客户拥有保额为：" + i*1000).getBytes(RemotingHelper.DEFAULT_CHARSET));
            // 异步启一个线程去做操作
            producer.send(msg, new SendCallback() {
                @Override
                public void onSuccess(SendResult sendResult) {
                    log.info("发送成功了！，返回消息为：{}", sendResult);
                }

                @Override
                public void onException(Throwable throwable) {
                    log.info("发送失败了！原因：{}", throwable);
                }
            });
        }

        // 沉睡一会，等待所有线程发送消息完毕，否则producer提前关闭导致消息发不出去
        Thread.sleep(3000);

        producer.shutdown();
        log.info("所有消息发送完毕！");
    }

}
